import indi.hjhk.csv.CSVReader;
import indi.hjhk.log.Logger;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.session.Session;
import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;

import java.io.IOException;
import java.util.ArrayList;

public class IotdbCarDataTrans {
    public static void main(String[] args) throws CSVReader.CSVFormatException, IOException {
        Logger logger = Logger.getPureLogger("iotdb");
        logger.disableAutoFlush();

        long totalSuccess = 0, totalFailed = 0, totalCsvElapsed = 0, totalIotdbElapsed = 0;

        ArrayList<String> csvFiles = new ArrayList<>();
        csvFiles.add("/hdd/users/hjh/test.csv");

        Session session = new Session.Builder().build();
        try {
            session.open();
        } catch (IoTDBConnectionException e) {
            throw new RuntimeException(e);
        }

        String databaseName = "root.car";

        try {
            session.createDatabase(databaseName);
        } catch (StatementExecutionException e) {
            if (e.getStatusCode() != TSStatusCode.DATABASE_ALREADY_EXISTS.getStatusCode()) {
                try {
                    session.close();
                } catch (IoTDBConnectionException ignored) {
                }
                throw new RuntimeException(e);
            }
        } catch (IoTDBConnectionException e) {
            try {
                session.close();
            } catch (IoTDBConnectionException ignored) {
            }
            throw new RuntimeException(e);
        }

        ArrayList<String> measurements = new ArrayList<>();
        ArrayList<TSDataType> datatypes = new ArrayList<>();
        ArrayList<Object> values = new ArrayList<>();

        measurements.add("stationId");
        datatypes.add(TSDataType.TEXT);

        measurements.add("passId");
        datatypes.add(TSDataType.TEXT);

        final int commitSize = 1000, statSize = 300000;
        logger.logOnStdout("commit size= %d, stat size= %d", commitSize, statSize);

        for (String csvFilePath : csvFiles) {
            long offset = 0, success = 0, failed = 0;

            CSVReader csvReader = new CSVReader(csvFilePath);
            csvReader.readHeader();

            long startTime, csvElapsed = 0, IotdbElapsed = 0;
            long subCsvElapsed = 0, subIotdbElapsed = 0;

            logger.logOnStdout("handling data within file %s:", csvFilePath);

            int successCount = 0, failedCount = 0;
            try {
                while (csvReader.next()) {
                    startTime = System.currentTimeMillis();
                    String carId = csvReader.getString("VLP") + "_" + csvReader.getString("VLPC");
                    long timestamp = csvReader.getLong("TIME");
                    String passId = csvReader.getString("PASSID");
                    String stationId = csvReader.getString("STATIONID");

                    subCsvElapsed += System.currentTimeMillis() - startTime;
                    startTime = System.currentTimeMillis();

                    String deviceId = databaseName+"."+carId;
                    if (!session.checkTimeseriesExists(deviceId+".stationId")){
                        session.createTimeseries(deviceId+".stationId", TSDataType.TEXT, TSEncoding.PLAIN, CompressionType.SNAPPY);
                        session.createTimeseries(deviceId+".passId", TSDataType.TEXT, TSEncoding.PLAIN, CompressionType.SNAPPY);
                    }
                    values.clear();
                    values.add(stationId);
                    values.add(passId);
                    session.insertRecord(deviceId, timestamp, measurements, datatypes, values);

                    successCount++;

                    subIotdbElapsed += System.currentTimeMillis() - startTime;

                    if (successCount % commitSize == 0){
                        session.close();
                        session.open();
                    }

                    if (successCount >= statSize){
                        csvElapsed += subCsvElapsed;
                        IotdbElapsed += subIotdbElapsed;

                        success += successCount;
                        failed += failedCount;
                        offset += successCount + failedCount;

                        logger.logOnStdout("\tsuccessfully committed %d data (failed: %d, end index= %d), timeSpent: CSVReader: %dms + Iotdb: %dms = %dms", successCount, failedCount, offset - 1, subCsvElapsed, subIotdbElapsed, subCsvElapsed + subIotdbElapsed);

                        successCount = 0;
                        failedCount = 0;
                        subCsvElapsed = 0;
                        subIotdbElapsed = 0;
                    }
                }

                csvElapsed += subCsvElapsed;
                IotdbElapsed += subIotdbElapsed;

                success += successCount;
                failed += failedCount;
                offset += successCount + failedCount;

                logger.logOnStdout("\tsuccessfully committed %d data (failed: %d, end index= %d), timeSpent: CSVReader: %dms + Iotdb: %dms = %dms", successCount, failedCount, offset - 1, subCsvElapsed, subIotdbElapsed, subCsvElapsed + subIotdbElapsed);

                totalCsvElapsed += csvElapsed;
                totalIotdbElapsed += IotdbElapsed;
                totalSuccess += success;
                totalFailed += failed;

            } catch (IoTDBConnectionException | StatementExecutionException e) {
                try {
                    session.close();
                } catch (IoTDBConnectionException ignored) {
                }
                logger.logOnStderr("Problem occurred when transferring from csv file %s to ArcadeDB at offset %d (+%d)", csvFilePath, offset, successCount);
                logger.flush();
                throw new RuntimeException(e);
            }

            logger.logOnStdout(String.format("Transferred from csv file %s to IoTDB(success: %d, failed: %d), CSV: %dms + IoTDB: %dms = %dms", csvFilePath, success, failed, csvElapsed, IotdbElapsed, csvElapsed + IotdbElapsed));
        }
        logger.logOnStdout(String.format("Total: Transferred from %d csv file(success: %d, failed: %d) to IoTDB, CSV: %dms + IoTDB: %dms = %dms", csvFiles.size(), totalSuccess, totalFailed, totalCsvElapsed, totalIotdbElapsed, totalCsvElapsed + totalIotdbElapsed));
        try {
            session.close();
        } catch (IoTDBConnectionException ignored) {
        }
        logger.flush();
    }
}
